package com.yet.container.registry.redis;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.Ordered;
import com.alibaba.fastjson.JSON;
import com.yet.container.registry.RegistryCenter;
import com.yet.container.registry.RegistryMeta;
import com.yet.container.registry.RegistryMeta.Address;
import com.yet.container.registry.RegistryMeta.ServiceMeta;
import com.yet.listener.RegistryEvent;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

public class RedisRegistry implements RegistryCenter {

	public static final Logger logger = LoggerFactory.getLogger(RedisRegistry.class);

	private JedisOperater operater;

	private final String name;

	private final String rootPath;

	private ApplicationEventPublisher publisher;

	public final YetJedisPubSub yetJedisPubSub = new YetJedisPubSub();

	private volatile State state = new State("Redis注册中心", true);

	public static final String REGISTRY_LISTENER_CHANNEL = "registry_listener_channel";

	public static final String UNREGISTRY_LISTENER_CHANNEL = "unregistry_listener_channel";

	private static final Map<String, RegistryEvent.Type> types = new HashMap<>(4);

	private static final MessageFormat MESSAGE_FORMAT = new MessageFormat("[name:{0}, rootPath:{1}, address:{2}]");

	static {
		types.put(REGISTRY_LISTENER_CHANNEL, RegistryEvent.Type.REGISTRY);
		types.put(UNREGISTRY_LISTENER_CHANNEL, RegistryEvent.Type.UNREGISTRY);
	}

	public static final ExecutorService pool = Executors.newFixedThreadPool(4, new ThreadFactory() {
		@Override
		public Thread newThread(Runnable r) {
			return new Thread(r, "redis-listener");
		}
	});

	public RedisRegistry(String name, String rootPath, JedisOperater operater) {
		this.operater = operater;
		this.name = name;
		this.rootPath = rootPath;
	}

	public RedisRegistry(String name, String rootPath) {
		super();
		this.name = name;
		this.rootPath = rootPath;
	}

	/**
	 * sadd REGISTRY_PREFIX-name-version
	 */
	@Override
	public void register(RegistryMeta meta) {
		String key = path(meta.getServiceMeta());
		operater.getJedisAndConsumer(j -> {
			String addressStr = j.hget(rootPath, key);
			List<Address> addressList;
			if (StringUtils.isBlank(addressStr)) {
				addressList = new ArrayList<>();
			} else {
				addressList = JSON.parseArray(addressStr, Address.class);
			}
			if (!addressList.contains(meta.getAddress())) {
				addressList.add(meta.getAddress());
				j.publish(REGISTRY_LISTENER_CHANNEL, JSON.toJSONString(meta.getServiceMeta()));
			}
			j.hset(rootPath, key, JSON.toJSONString(addressList));
		});
	}

	@Override
	public void unregister(RegistryMeta meta) {
		String key = path(meta.getServiceMeta());
		operater.getJedisAndConsumer(j -> {
			String addressStr = j.hget(rootPath, key);
			if (StringUtils.isBlank(addressStr)) {
				return;
			}
			List<Address> addressList = JSON.parseArray(addressStr, Address.class);
			addressList.remove(meta.getAddress());
			j.hset(rootPath, key, JSON.toJSONString(addressList));
			j.publish(UNREGISTRY_LISTENER_CHANNEL, JSON.toJSONString(meta.getServiceMeta()));
		});
	}

	@Override
	public void subscribe(final ServiceMeta serviceMeta) {
		Jedis j = operater.getJedis();
		pool.execute(() -> {
			try {
				j.psubscribe(yetJedisPubSub, REGISTRY_LISTENER_CHANNEL, UNREGISTRY_LISTENER_CHANNEL);
			} finally {
				j.close();
			}
		});
	}

	@Override
	public void unsubscribe(ServiceMeta serviceMeta) {
		yetJedisPubSub.unsubscribe(REGISTRY_LISTENER_CHANNEL, UNREGISTRY_LISTENER_CHANNEL);
	}

	@Override
	public List<Address> lookup(ServiceMeta serviceMeta) {
		String key = path(serviceMeta);
		return operater.getJedisAndFunction(j -> {
			return JSON.parseArray(j.hget(rootPath, key), Address.class);
		});
	}

	@Override
	public String getName() {
		return name;
	}

	@Override
	public void start() throws RuntimeException {
		operater.start();
	}

	@Override
	public void stop() {
		if (operater != null) {
			operater.close();
		}
		state.setStarted(false);
	}

	public class YetJedisPubSub extends JedisPubSub {

		@Override
		public void onPMessage(String pattern, String channel, String message) {
			publisher.publishEvent(
					new RegistryEvent(getName(), JSON.parseObject(message, ServiceMeta.class), types.get(pattern)));
		}

	}

	private static String path(String... strs) {
		return joinWith("-", strs);
	}

	private static String path(ServiceMeta serviceMeta) {
		return path(serviceMeta.getName(), serviceMeta.getVersion());
	}

	private static String joinWith(String marker, String... strs) {
		return String.join(marker, strs);
	}

	@Override
	public Map<ServiceMeta, List<Address>> load() {
		Map<ServiceMeta, List<Address>> cache = new HashMap<>();
		String key = path(rootPath);
		operater.getJedisAndConsumer(j -> {
			Map<String, String> metas = j.hgetAll(key);
			metas.forEach((k, v) -> {
				cache.put(getServiceMetaFromRedisKey(k), JSON.parseArray(v, Address.class));
			});
		});
		return Collections.unmodifiableMap(cache);
	}

	public static ServiceMeta getServiceMetaFromRedisKey(String key) {
		String[] split = key.split("-");
		if (split.length == 2) {
			return new ServiceMeta(split[0], split[1]);
		}
		throw new IllegalStateException("错误的键名:" + key);
	}

	public void setOperater(JedisOperater operater) {
		this.operater = operater;
	}

	@Override
	public int getOrder() {
		return Ordered.HIGHEST_PRECEDENCE + 100;
	}

	public ApplicationEventPublisher getPublisher() {
		return publisher;
	}

	public void setPublisher(ApplicationEventPublisher publisher) {
		this.publisher = publisher;
	}

	public JedisOperater getOperater() {
		return operater;
	}

	public String getRootPath() {
		return rootPath;
	}

	@Override
	public State getState() {
		state.setInfo(MESSAGE_FORMAT.format(new Object[] { getName(), getRootPath(), getOperater().getUri() }));
		return state;
	}

}
